package com.leilei;

import cn.hutool.core.util.RandomUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Random;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.UNBOUNDED_ROW;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/21 14:35
 * @desc 基于滚动时间窗口，计算每段时间报警数量
 */
public class Flink_SQL_4_TumbleWindow {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Location> locationSource = env.addSource(new LocationSource());
        env.setParallelism(1);
        //设置water-maker水位线
        DataStream<Location> finalLocationSource = locationSource.
                assignTimestampsAndWatermarks(WatermarkStrategy.
                        <Location>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, timestamp) -> event.getDevTime()));
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

  tableEnv.createTemporaryView("locationSource",finalLocationSource, $("vehicleId"), $("plate"),
                $("color"), $("date"), $("gpsSpeed"), $("limitSpeed"), $("devTime").rowtime());
        String sql = "select " +
                "plate," +
                "max(gpsSpeed) as maxGpsSpeed, " +
                "count(1) as totalAlarm " +
                "from locationSource where (gpsSpeed > limitSpeed) " +
                "group by plate," +
                "tumble(devTime, interval '5' second) ";
        Table resultTable = tableEnv.sqlQuery(sql);
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnv.toRetractStream(resultTable, Row.class);
        tuple2DataStream.filter(t->t.f0).print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }







    public static class LocationSource implements SourceFunction<Location> {
        Boolean flag = true;

        @Override
        public void run(SourceContext<Location> ctx) throws Exception {
            Random random = new Random();
            while (flag) {
                int vehicleId = random.nextInt(5);
                Location location = Location.builder()
                        .vehicleId(vehicleId)
                        .plate("川A000" + vehicleId)
                        .color("黄")
                        .date(Integer.parseInt(LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE)))
                        .gpsSpeed(RandomUtil.randomInt(90, 100))
                        .limitSpeed(RandomUtil.randomInt(88, 95))
                        .devTime(System.currentTimeMillis() - RandomUtil.randomInt(3,10) * 1000)
                        .build();
                ctx.collect(location);
                Thread.sleep(1000);

            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }
}
